package com.tj.bigdata.syncblock.strategy;

import com.tj.bigdata.syncblock.config.Web3jConfig;
import com.tj.bigdata.syncblock.entity.BlockInfo;
import com.tj.bigdata.syncblock.entity.ReceiptLog;
import com.tj.bigdata.syncblock.entity.Transactions;
import com.tj.bigdata.syncblock.producer.BlockChainInfoKafkaSender;
import com.tj.bigdata.syncblock.service.EsUpService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.DefaultBlockParameter;
import org.web3j.protocol.core.methods.response.EthBlock;
import org.web3j.protocol.core.methods.response.TransactionReceipt;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 *
 *
 * @author guoch
 */
@Slf4j
@Service
public class Web3jKafkaSyncImpl implements SyncBlockStrategy {

    @Autowired
    private EsUpService esUpService;

    @Autowired
    private ThreadPoolExecutor execute;

    @Autowired
    private Web3jConfig web3jConfig;

    @Autowired
    private BlockChainInfoKafkaSender blockChainInfoKafkaSender;

    @Override
    public void batchBlockchain() {
        Web3j web3j = web3jConfig.getWeb3j();
        log.debug("连接以太坊成功");
        BigInteger maxBlockNumber = esUpService.maxBlockBumber() != null ? esUpService.maxBlockBumber().subtract(new BigInteger(String.valueOf(2000))) : BigInteger.ZERO;
        web3j.replayPastAndFutureBlocksFlowable(DefaultBlockParameter.valueOf(maxBlockNumber), true).retry(3)
                .doOnError(e -> log.error("on error:{}", e.getMessage()))
                .subscribe(this::executeBlock);
    }

    @Override
    public void batchBlockchain(BigInteger begin, BigInteger end) {

    }

    public void execute(Runnable r) {
        execute.execute(r);
//        CompletableFuture.runAsync(r,execute);
    }

    public void executeBlock(EthBlock block) {

        execute(() -> {
            // 获取到所需的块信息
            EthBlock.Block ethBlock = block.getBlock();
            BlockInfo blockInfo = new BlockInfo();
            BeanUtils.copyProperties(ethBlock, blockInfo);
            log.info("==========区块信息封装成功 {}====》", ethBlock.getNumber());
            blockChainInfoKafkaSender.sendBlock(blockInfo);
            // transaction信息获取
            executeTransaction(ethBlock.getTransactions());
            log.info("==========区块信息发送kafka成功 {}=======", ethBlock.getNumber());
        });
    }

    public void executeTransaction(List<EthBlock.TransactionResult> transactions) {

        if (transactions.size() == 0) {
            return;
        }
        try {
            Web3j web3j = web3jConfig.getWeb3j();
            Transactions transactionss;
            for (EthBlock.TransactionResult<EthBlock.TransactionObject> transactionResult : transactions) {
                EthBlock.TransactionObject transaction = transactionResult.get();
                transactionss = new Transactions();
                transactionss.setBlockHash(transaction.getBlockHash()).setBlockNumber(transaction.getBlockNumber())
                        .setFrom(transaction.getFrom()).setTo(transaction.getTo()).setGas(transaction.getGas()).setGasPrice(transaction.getGasPrice())
                        .setHash(transaction.getHash()).setTransactionIndex(transaction.getTransactionIndex()).setInput(transaction.getInput()).setNonce(transaction.getNonce())
                        .setValue(transaction.getValue()).setV(transaction.getV()).setR(transaction.getR()).setS(transaction.getS());
                TransactionReceipt receipt = web3j.ethGetTransactionReceipt(transaction.getHash()).send().getResult();
                if (receipt != null) {
                    BeanUtils.copyProperties(receipt, transactionss);
                    ArrayList<ReceiptLog> receiptLogs = new ArrayList<>();
                    receipt.getLogs().forEach(log1 -> {
                        ReceiptLog receiptLog = new ReceiptLog();
                        receiptLog.setRemoved(log1.isRemoved()).setLogIndex(log1.getLogIndex()).setType(log1.getType())
                                .setAddress(log1.getAddress()).setData(log1.getData()).setTopics(log1.getTopics());
                        receiptLogs.add(receiptLog);
                    });
                    transactionss.setLogs(receiptLogs);
                }
                blockChainInfoKafkaSender.sendTransaction(transactionss);
                log.info("==========交易信息发送kafka成功 {}=======", transactionss.getHash());
            }
        } catch (IOException e) {
            log.error("transaction input error, msg:{}", e.getMessage());
        }
    }


    @PreDestroy
    public void preDestroy() {
        log.info("大数据 - 区块同步 - 结束");
    }
}
